package com.fowo.api.sys.cns.listeners;

import com.fowo.api.sys.cns.BatchChangeNoticeListener;
import com.fowo.api.sys.cns.ChangeNoticeData;
import com.fowo.api.sys.cns.ChangeNoticeListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 用于收集普通通知转换为批量通知
 * @author yl_ls
 */
@Slf4j
@Component
public class ToBatchChangeNoticeListener implements ChangeNoticeListener {
    /**
     * 最大批次记录数
     */
    public static final int MAX_SEND_BATCH_RECORDS = 100;
    /**
     * 最大收到消息到发送间的延时
     */
    public static final int MAX_SEND_BATCH_DELAY = 5000;

    @Resource
    private Collection<BatchChangeNoticeListener> listeners;

    private final Queue<ChangeNoticeData> changeNoticeDataQueue = new ConcurrentLinkedQueue<>();
    private ScheduledThreadPoolExecutor poolExecutor;
    private ScheduledFuture<?> noticeSchedule;

    @PostConstruct
    public void init() {
        ScheduledThreadPoolExecutor poolExecutor = new ScheduledThreadPoolExecutor(1);
        poolExecutor.setMaximumPoolSize(1);
        this.poolExecutor = poolExecutor;
    }

    @Override
    public synchronized void notice(ChangeNoticeData noticeData) {
        changeNoticeDataQueue.add(noticeData);
        afterNotice();
    }

    private void afterNotice() {
        if(changeNoticeDataQueue.size() >= MAX_SEND_BATCH_RECORDS) {
            if (noticeSchedule != null && !noticeSchedule.isCancelled()) {
                noticeSchedule.cancel(false);
                noticeSchedule = null;
            }
            startNotices();
        } else {
            if (noticeSchedule == null) {
                noticeSchedule = poolExecutor.schedule(this::startNotices, MAX_SEND_BATCH_DELAY, TimeUnit.MILLISECONDS);
            }
        }
    }

    private void startNotices() {
        List<ChangeNoticeData> executeQueue = new ArrayList<>();
        while (!changeNoticeDataQueue.isEmpty()) {
            executeQueue.add(changeNoticeDataQueue.remove());
        }
        poolExecutor.execute(() -> handleNotices(executeQueue));
    }

    private void handleNotices(List<ChangeNoticeData> changeNoticeDataList) {
        List<ChangeNoticeData> mixChangeNoticeDataList = createMixChangeNoticeDataList(changeNoticeDataList);
        for (BatchChangeNoticeListener listener : listeners) {
            try {
                listener.batchNotice(changeNoticeDataList, mixChangeNoticeDataList);
            } catch (Exception e) {
                log.error("变更通知发送给 {} 时发生错误", listener, e);
            }
        }
    }

    private List<ChangeNoticeData> createMixChangeNoticeDataList(List<ChangeNoticeData> changeNoticeDataList) {
        List<ChangeNoticeData> mixList = new ArrayList<>(changeNoticeDataList.size());
        Map<String, Integer> objectNoticeMap = new HashMap<>(changeNoticeDataList.size());
        for (ChangeNoticeData noticeData : changeNoticeDataList) {
            if(!objectNoticeMap.containsKey(noticeData.getObjectName())) {
                objectNoticeMap.put(noticeData.getObjectName(), mixList.size());
                mixList.add(noticeData);
                continue;
            }
            Integer index = objectNoticeMap.get(noticeData.getObjectName());
            ChangeNoticeData prevNoticeData = mixList.get(index);
            ChangeNoticeData newNoticeData = mixNotices(prevNoticeData, noticeData);
            mixList.set(index, newNoticeData);
        }
        return mixList;
    }

    private ChangeNoticeData mixNotices(ChangeNoticeData prevNoticeData, ChangeNoticeData noticeData) {
        ChangeNoticeData newNoticeData = new ChangeNoticeData();
        if (noticeData.getType() == ChangeNoticeData.TYPE_BATCH_ALL) {
            return newNoticeData;
        }
        newNoticeData.setInsertKeys(mixKeys(prevNoticeData.getInsertKeys(), noticeData.getInsertKeys()));
        newNoticeData.setUpdateKeys(mixKeys(prevNoticeData.getUpdateKeys(), noticeData.getUpdateKeys()));
        newNoticeData.setInsertKeys(mixKeys(prevNoticeData.getRemoveKeys(), noticeData.getRemoveKeys()));
        return null;
    }

    private List<Long> mixKeys(List<Long> prevKeys, List<Long> keys) {
        if (prevKeys == null || prevKeys.isEmpty()) {
            return keys;
        }
        if (keys == null || keys.isEmpty()) {
            return prevKeys;
        }
        Set<Long> newKeys = new HashSet<>(prevKeys.size() + keys.size());
        newKeys.addAll(prevKeys);
        newKeys.addAll(keys);
        return new ArrayList<>(newKeys);
    }
}
